package a_aa_amainpackage.a_o2odata_deal.utils

import a_aa_amainpackage.a_o2odata_deal.config.config._
import org.apache.spark.sql.{DataFrame, SQLContext}

/*
*项目名: o2odata_commend
*文件名: sell_deal
*创建者: hancongcong
*创建时间:2019/8/12 9:48
*描述: TODO
*/ object sell_deal {


  /**
    * 销量计算
    * @param sqlContext
    * @param table
    * @param currentMonthSourceSellPath
    * @param currentMonthBucaiSellPath
    * @param lastMonthSourceSellPath
    * @return
    * Just Looking down running==Pointless grievances
    */
  def tmall_sell_clear(sqlContext: SQLContext): DataFrame = {
    //天猫原始销量
    /*val sourceSellData = sqlContext.read.json(s"${source_obs_data_path}").where("BC_type='B'").select("good_id", "commentCount", "coupon", "crawl_date", "priceText", "preSaleCount", "price_desc", "promotion_price", "sellCount"
      , "categoryId", "brandValueId", "brandName", "shopId", "title", "Base_Info", "detailCategoryName","extendedData")
*/

    val sourceSellData = sqlContext.read.json(s"${source_obs_data_path}").where("BC_type='B'").select("good_id", "commentCount", "coupon", "crawl_date", "priceText", "preSaleCount", "price_desc", "promotion_price", "sellCount"
      , "categoryId", "brandValueId", "brandName", "shopId", "title", "Base_Info", "detailCategoryName","extendedData")

    sourceSellData.registerTempTable("sourceSellData")
    //天猫补采销量
    //val addSellData = sqlContext.read.json(s"${totalsell_path}")
    //.registerTempTable("addSellData")
    //天猫上月销量
    val tmlastMonth = sqlContext.read.json(s"${last_source_obs_data_path}").select("good_id", "commentCount")
    tmlastMonth.registerTempTable("tmlastMonth")

    //////////////////////////临时代码  减去预售销量
    /*val yuhshou_data = sqlContext.read.json("s3a://o2o-dataproces-group/panzonghao/last_yushou_month11_result/")
    yuhshou_data.createOrReplaceTempView("yuhshou_data")
    sqlContext.sql(
      s"""
         |
       """.stripMargin)*/



    //分本月数据 大于0(等于0过滤掉)
    val sell1 = sourceSellData.where("sellCount > 0")
    sell1.registerTempTable("sell1")
    //小于0
    // 当tmall_detail_1901集合里商品销量为-1时的处理流程：
    //1.首先去tmall_id_from_brand_api_1901集合关联，关联到的就用对应的sellCont(月销量)
    val sell2 = sourceSellData.where("sellCount < 0")
    sell2.registerTempTable("sell2")

    //val api_sell = sqlContext.read.json(s"${brandpath}")
    //api_sell.registerTempTable("api_sell")
    //val sell0 = sqlContext.sql(
    //  """
    //    |select
    //    |t1.*,
    //    |t2.sellCount sells
    //    |from
    //    |sell2 t1
    //    |join
    //    |api_sell t2
    //    |on t1.good_id = t2.good_id
    //  """.stripMargin).drop("sellCount").withColumnRenamed("sells", "sellCount")
    //  .where("sellCount > 0")
    ////关联不到的
    //val sell0nojoin = sqlContext.sql(
    //  """
    //    |select
    //    |t1.*
    //    |from
    //    |sell2 t1
    //    |left join
    //    |api_sell t2
    //    |on t1.good_id = t2.good_id
    //    |where t2.good_id is null
    //  """.stripMargin)
    //sell0nojoin.registerTempTable("sell0nojoin")
    //小于0的数据与补采的关联 关联上的 销量 = 补采add_to_field字段最后一个值-第一个值
    //val sellbu = sqlContext.sql(
    //  """
    //    |select
    //    |t1.*,
    //    |(t2.add_to_field[size(add_to_field)-1].totalSell - t2.add_to_field[0].totalSell) sells
    //    |from
    //    |sell0nojoin t1
    //    |join
    //    |addSellData t2
    //    |on t1.good_id = t2.good_id
    //  """.stripMargin).drop("sellCount").withColumnRenamed("sells", "sellCount").where("sellCount > 0")
    ////关联不上的
    //val nosell = sqlContext.sql(
    //  """
    //    |select
    //    |t1.*
    //    |from
    //    |sell0nojoin t1
    //    |left join
    //    |addSellData t2
    //    |on t1.good_id = t2.good_id
    //    |where t2.good_id is null
    //  """.stripMargin)
    //nosell.registerTempTable("nosell")
    //与上月数据关联 关联上的 销量=当月评论-上月评论(结果取大于0)
    // val noselljoin = sqlContext.sql(
    //   """
    //     |select
    //     |t1.*,
    //     |(t1.commentCount-t2.commentCount) sells
    //     |from
    //     |nosell t1
    //     |join
    //     |tmlastMonth t2
    //     |on t1.good_id=t2.good_id
    //   """.stripMargin).drop("sellCount").withColumnRenamed("sells", "sellCount").where("sellCount > 0")


    val noselljoin = sqlContext.sql(
      """
        |select
        |t1.*,
        |(t1.commentCount-t2.commentCount) sells
        |from
        |sell2 t1
        |join
        |tmlastMonth t2
        |on t1.good_id=t2.good_id
      """.stripMargin).drop("sellCount").withColumnRenamed("sells", "sellCount").where("sellCount > 0")


    //合并： 销量 = sellbu+noselljoin+sell1
    /*val value: DataFrame = sellbu.select("good_id", "sellCount", "shopId", "coupon", "priceText")
      .union(noselljoin.select("good_id", "sellCount", "shopId", "coupon", "priceText"))
      .union(sell0.select("good_id", "sellCount", "shopId", "coupon", "priceText"))
      .union(sell1.select("good_id", "sellCount", "shopId", "coupon", "priceText"))
      .toDF("good_id", "sellCount", "shopId", "coupon", "priceText")
      .dropDuplicates("good_id")
    value.registerTempTable("value")*/

    val value: DataFrame = sell1.selectExpr("good_id", "sellCount", "shopId", "coupon", "priceText","extendedData")
      .union(noselljoin.selectExpr("good_id", "sellCount", "shopId", "coupon", "priceText","extendedData"))
      //.toDF("good_id", "sellCount", "shopId", "coupon", "priceText","extendedData")
      .dropDuplicates("good_id")
    value.registerTempTable("value")
    //转换类型
    val sellResult_df = sqlContext.sql("select *,CAST(sellCount as bigint) sellCounts from value").drop("sellCount").withColumnRenamed("sellCounts", "sellCount")
    return sellResult_df
  }

}
